package com.practice.util;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.ObjectUtil;
import lombok.Builder;
import lombok.extern.slf4j.Slf4j;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

/**
 * 调度线程处理器
 * @param <R> 元素
 * @param <T> 元素id
 * @author: kayn
 * 2025-01-17 21:11
 */
@Slf4j
@Builder
public class ExecutePoolHandler<T,R>{


    /**
     * 调度线程执行方法
     */
    private Consumer<R> execute;

    /**
     * 获取元素id
     */
    private Function<R,T> id;

    /**
     * 获取全部元素
     */
    private Supplier<List<R>> all;

    /**
     * 获取执行间隔
     */
    private Function<R,Integer> interval;

    /**
     * 调度线程类别
     */
    private String executeEnum;

    /**
     * 默认检测间隔
     */
    @Builder.Default
    private Integer detectInterval = 1000;

    /**
     * 默认执行间隔
     */
    @Builder.Default
    private Integer executeInterval = 500;

    /**
     * 元素列表
     */
    @Builder.Default
    public List<R> elementList = CollUtil.newArrayList();

    /**
     * Executors存储池
     */
    @Builder.Default
    public final ConcurrentHashMapReference<T, ScheduledExecutorService> executorPool = new ConcurrentHashMapReference<>();

    /**
     * 启动处理器核心
     */
    public void start() {

        ThreadFactory factory = Thread.ofVirtual().factory();

        ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(factory);

        executor.scheduleAtFixedRate(()->this.loadThread(factory), 0, detectInterval, TimeUnit.MILLISECONDS);

        log.info("{}检测线程已启动", executeEnum);
    }

    /**
     * 加载执行线程
     */
    private void loadThread(ThreadFactory factory) {

        try{

            List<R> elementList = all.get();
            List<T> elementIdList = elementList.stream().map(id).toList();

            List<R> newElementList = elementList.stream().filter(r -> !executorPool.containsKey(id.apply(r))).toList();
            newElementList.forEach((r)->this.startThread(r,factory));
            this.elementList.addAll(newElementList);

            executorPool.keySet().stream().filter(id -> !elementIdList.contains(id)).toList().forEach(this::clearExecutor);
        }catch (Exception e) {
            log.error("【{}线程调度执行器】加载线程异常", executeEnum,e);
        }
    }

    /**
     * 启动执行线程
     */
    private void startThread(R r, ThreadFactory factory) {

        try {

            ScheduledExecutorService executor = executorPool.get(id.apply(r), () -> Executors.newSingleThreadScheduledExecutor(factory));

            Integer executeInterval = ObjectUtil.isNotNull(interval.apply(r)) ? interval.apply(r) : this.executeInterval;

            executor.scheduleWithFixedDelay(() -> execute.accept(r), 0, executeInterval, TimeUnit.MILLISECONDS);

        }catch (Exception e){
            log.error("【{}线程调度执行器】调度线程异常",executeEnum,e);
        }
    }

    /**
     * 单个清除Executor
     * @param id 线程池id
     */
    public void clearExecutor(T id) {
        try {
            AtomicReference<ScheduledExecutorService> reference = executorPool.get(id);
            if (ObjectUtil.isNotNull(reference)) {
                ScheduledExecutorService executorService = reference.get();
                if (executorService != null) {
                    executorService.shutdownNow();
                }
            }
            executorPool.remove(id);
        }catch (Exception e) {
            log.error("【{}线程调度执行器】清理线程池失败",executeEnum,e);
        }
    }
}
